package com.example.java_message_queue.mqserver.datacenter;

import com.example.java_message_queue.common.MqException;
import com.example.java_message_queue.mqserver.core.Binding;
import com.example.java_message_queue.mqserver.core.Exchange;
import com.example.java_message_queue.mqserver.core.MSGQueue;
import com.example.java_message_queue.mqserver.core.Message;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

// 管理所有的内存数据.
public class MemoryDataCenter {
    // key 是 exchangeName
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    // key 是 queueName
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    // 第一个 key 是 exchangeName, 第二个 key 是 queueName
    private ConcurrentHashMap<String, HashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    // 保存所有消息, key 是 messageId
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    // key 是 queueName
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    // 用来存放待确认的消息
    // key1 是 queueName, key2 是 messageId.
    // 这个结构不需要有对应的硬盘数据. 换句话说, 如果某个消息消费了, 但是没有 ack, 这个时候 broker 宕机了, 那么重启 broker 之后
    // 就把刚才的消息当做从来没消费过.
    private ConcurrentHashMap<String, HashMap<String, Message>> queueMessageWaitAck = new ConcurrentHashMap<>();

    public void init() {
    }

    // 从硬盘上恢复数据
    public void recovery(DiskDataCenter diskDataCenter) throws MqException, IOException, ClassNotFoundException {
        // 1. 恢复交换机数据
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for (Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(), exchange);
        }
        // 2. 恢复队列数据
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for (MSGQueue queue : queues) {
            queueMap.put(queue.getName(), queue);
        }
        // 3. 恢复绑定数据
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings) {
            HashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), k -> new HashMap<>());
            bindingMap.put(binding.getQueueName(), binding);
        }
        // 4. 恢复消息数据
        //    只需要恢复 queueMessageMap 和 messageMap
        //    queueMessageWaitAck 则不必恢复. 未被确认的消息只是在内存存储. 如果这个时候 broker 宕机了, 则消息视为没有被消费过.
        for (MSGQueue queue : queues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(), messages);
            for (Message message : messages) {
                messageMap.put(message.getMessageId(), message);
            }
        }
    }

    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
    }

    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
    }

    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
    }

    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
    }

    public void insertBinding(Binding binding) throws MqException {
        HashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(), k -> new HashMap<>());
        synchronized (bindingMap) {
            // 不存在就创建一份
            if (bindingMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter] 绑定已经存在! exchangeName=" + binding.getExchangeName()
                        + ", queueName=" + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
        }
    }

    public Binding getBinding(String queueName, String exchangeName) {
        HashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        synchronized (bindingMap) {
            return bindingMap.get(queueName);
        }
    }

    public void deleteBinding(Binding binding) throws MqException {
        HashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            throw new MqException("[MemoryDataCenter] 绑定不存在! exchangeName=" + binding.getExchangeName()
                    + ", queueName=" + binding.getQueueName());
        }
        synchronized (bindingMap) {
            Binding toDelete = bindingMap.get(binding.getQueueName());
            if (toDelete == null) {
                throw new MqException("[MemoryDataCenter] 绑定不存在! exchangeName=" + binding.getExchangeName()
                        + ", queueName=" + binding.getQueueName());
            }
            bindingMap.remove(binding.getQueueName());
        }
    }

    public Map<String, Binding> getBindingsByExchange(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    // 查询指定的消息
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    // 向消息中心中添加消息
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        System.out.println("[MemoryCenter] 新消息被添加! messageId=" + message.getMessageId());
    }

    // 从消息中心删除消息
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("[MemoryCenter] 消息被彻底删除! messageId=" + messageId);
    }

    // 发送消息到指定队列中
    public void sendMessage(MSGQueue queue, Message message) {
        List<Message> messageList = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());
        synchronized (messageList) {
            messageList.add(message);
        }
        // 如果消息已经存在, 重复调用也没啥大不了的.
        addMessage(message);
        System.out.println("[MemoryCenter] 消息被投递到队列中! messageId=" + message.getMessageId() + ", queueName=" + queue.getName());
    }

    public Message pollMessage(String queueName) throws MqException {
        List<Message> messageList = queueMessageMap.get(queueName);
        if (messageList == null) {
            throw new MqException("[MemoryDataCenter] 队列不存在! queueName=" + queueName);
        }
        synchronized (messageList) {
            if (messageList.size() == 0) {
                return null;
            }
            // 出队列头元素
            Message currentMessage = messageList.remove(0);
            System.out.println("[MemoryCenter] 消息从队列中取出! messageId=" + currentMessage.getMessageId() + ", queueName=" + queueName);
            return currentMessage;
        }
    }

    public int getMessageCount(String queueName) throws MqException {
        List<Message> messageList = queueMessageMap.get(queueName);
        if (messageList == null) {
            // 如果队列不存在, 则直接返回长度 0, 说明该 queueName 下还没有消息.
            return 0;
        }
        synchronized (messageList) {
            return messageList.size();
        }
    }

    // 未被确认的消息, 先临时存放一下
    public void addMessageWaitAck(String queueName, Message message) {
        HashMap<String, Message> messageHashMap = queueMessageWaitAck.computeIfAbsent(queueName, k -> new HashMap<>());
        synchronized (messageHashMap) {
            messageHashMap.put(message.getMessageId(), message);
        }
        System.out.println("[MemoryCenter] 消息进入待确认队列! messageId=" + message.getMessageId() + ", queueName=" + queueName);
    }

    // 消息被确认之后, 就可以真正删除了.
    public void removeMessageWaitAck(String queueName, String messageId) {
        HashMap<String, Message> messageHashMap = queueMessageWaitAck.get(queueName);
        if (messageHashMap == null) {
            return;
        }
        synchronized (messageHashMap) {
            messageHashMap.remove(messageId);
        }
        System.out.println("[MemoryCenter] 消息从待确认队列删除! messageId=" + messageId + ", queueName=" + queueName);
    }

    public Message getMessageWaitAck(String queueName, String messageId) {
        HashMap<String, Message> messageHashMap = queueMessageWaitAck.get(queueName);
        if (messageHashMap == null) {
            return null;
        }
        synchronized (messageHashMap) {
            return messageHashMap.get(messageId);
        }
    }
}
